#include<iostream>
#include<arpa/inet.h>
#include<sys/epoll.h>
#include<sys/socket.h>
#include<fcntl.h>
#include<cstring>
#include<unistd.h>
#include<atomic>
#include<thread>
#include<condition_variable>
#include"reactor.h"
#include"http.h"
using namespace std;

DistributeReactor::DistributeReactor():flag(true){
        listen_fd=socket(AF_INET,SOCK_STREAM | SOCK_NONBLOCK,0);
        if(listen_fd==-1)
        perror("listen socket error");

        epoll_fd=epoll_create1(1);
        if(epoll_fd==-1)
        perror("epoll create error:");
    }

DistributeReactor::~DistributeReactor(){
        close(listen_fd);
        close(epoll_fd);
    }
void DistributeReactor::BeginServer()
    {
        struct sockaddr_in addr;
        memset(&addr,0,sizeof(addr));
        addr.sin_family=AF_INET;
        addr.sin_port=htons(8080);
        addr.sin_addr.s_addr=htonl(INADDR_ANY);
        

        if(bind(listen_fd,(struct sockaddr *)&addr,sizeof(addr))==-1)
        perror("bind error");

        if(listen(listen_fd,SOMAXCONN)==-1)
        perror("listen error");

        cout<<"socket is listening on port "<<ntohs(addr.sin_port)<<endl;
        
        struct epoll_event event;
        event.events= EPOLLIN | EPOLLET;
        event.data.fd= listen_fd;
        if(epoll_ctl(epoll_fd,EPOLL_CTL_ADD,listen_fd,&event)==-1)
        perror("epoll ctl add listen_fd");

        struct epoll_event events[512]; //这里设置大小多少比较好呢？
        while(flag)
        {
            int nums=epoll_wait(epoll_fd,events,1024,-1);
            if(nums==-1)
            perror("epoll wait listen error ");
            for(int i=0;i<nums;i++)
            {
                struct sockaddr_in addr1;
                memset(&addr1,0,sizeof(addr1));
                socklen_t len = sizeof(addr1);
                int client_fd=accept(listen_fd,(struct sockaddr *)&addr1,&len);
                if(client_fd==-1)
                {
                    if(events[i].events & EAGAIN || events[i].events & EWOULDBLOCK)
                    continue;
                    else 
                    {
                        perror("accept error");
                        continue;
                    }
                }

                int flag = fcntl(client_fd, F_GETFL,0);
                if(flag==-1)
                {
                    perror("fcntl error ");
                    //优雅的错误处理
                    close(client_fd);
                    continue;
                }

                if(fcntl(client_fd,F_SETFL, flag | O_NONBLOCK)==-1)
                {
                    perror("fcntl set fd error");
                    close(client_fd);
                    continue;
                }
                
                //负载均衡策略分发给从reactor



            }
        }
    }

WorkReactor::WorkReactor():nums(0),flag(true)
    {
       epoll_fd=epoll_create1(0);
       if(epoll_fd==-1)
       {
        cout<<"WorkReactor: "<<this_thread::get_id()<<" ERROR\n";
        perror("work epoll create");
       }
    }
WorkReactor::~WorkReactor()
    {
        close(epoll_fd);
    }

void  WorkReactor::add(int client_fd,struct sockaddr_in & addr) 
    {

        struct epoll_event event;
        event.events = EPOLLIN | EPOLLRDHUP | EPOLLET;
        event.data.fd=client_fd;
        if(epoll_ctl(epoll_fd,EPOLL_CTL_ADD,client_fd,&event)==-1)
        {
            perror("epoll ctl add error work");
            close(client_fd);
        }

        char ip1[INET_ADDRSTRLEN];
        inet_ntop(AF_INET,&addr.sin_addr,ip1,INET_ADDRSTRLEN);
        string s("accept a connection from addr : ");
        s+=ip1;
        s+=+" port : "+ ntohs(addr.sin_port);
        write(1,s.c_str(),sizeof(s));
        nums++;
    }
   
void WorkReactor::run()
    {
        struct epoll_event events[512];
        while(flag)
        { 
            int n=epoll_wait(epoll_fd,events,512,-1);
            if(n==-1)
            {
                perror("epoll wait work error");
                break;
            } 
            for(int i=0;i<n;i++)
            {
               if(events[i].events & EPOLLRDHUP)
               {
                   //关闭链接
                   if(epoll_ctl(epoll_fd,EPOLL_CTL_DEL,events[i].data.fd,nullptr)==-1)
                   {
                    perror("epoll ctl del work");
                    continue;
                   }
                   close(events[i].data.fd);
                   continue;
               }

               if(events[i].events & EPOLLIN)
               {
                    char buffer[1024];
                    ssize_t readbytes;
                    do
                    {
                        readbytes=read(events[i].data.fd,buffer,1023);
                        if(readbytes==0)
                        {
                            cout<<"Connection closed by client fd:"<<events[i].data.fd<<endl;
                            if(epoll_ctl(epoll_fd,EPOLL_CTL_DEL,events[i].data.fd,nullptr)==-1)
                            {
                                perror("epollctl del failed");
                                break;
                            }
                            close(events[i].data.fd);
                            break;
                        }
                        else if(readbytes==-1)
                        {        
                                if(errno != EAGAIN && errno != EWOULDBLOCK )
                                {
                                    perror("Read ");
                                    if(epoll_ctl(epoll_fd,EPOLL_CTL_DEL,events[i].data.fd,nullptr)==-1)
                                    {
                                        perror("epollctl del failed");
                                        continue;
                                    }
                                    //关闭文件描述符
                                    close(events[i].data.fd);
                                    break;
                                }
                        }

                        buffer[readbytes] = '\0'; //字符串结束符
                        HandlHttp::handle(buffer,readbytes,events[i].data.fd);

                    }while(readbytes>0);
               }
            }
        }
    }
    
int WorkReactor::getnums()
    {
        return nums;
    }
